<?php namespace xlx;

use Workerman\Worker;
use Workerman\Connection\TcpConnection;
use Workerman\Connection\UdpConnection;
use Workerman\Events\EventInterface;
//\ini_set("error_log",__DIR__.'/err.log');
ini_set("memory_limit","3G");


class PrService
{

    ////////////// basic /////////////////
    static $reg_protocol = '\xlx\Protocols\Frame';
    static $reg_addr = '127.0.0.1:55046';
    static $reg_conn;//进程唯一 tcp接收



    static $client_data=['socket'=>[], 'iptable'=>[] ,'return'=>[] ];
    static $client_monitor=[];//
    static $client_exit=[];
    public static function connectServer()
    {
        if (isset(self::$reg_conn)) {
            return '';
        }
        if (empty(\Workerman\Worker::$globalEvent)) {
            \Workerman\Worker::$globalEvent = \xlx\Process::createEventLoop();
        }
        $reg_socket = stream_socket_client('tcp://'.self::$reg_addr, $errno, $errstr) or exit;
        $client_addr = stream_socket_get_name($reg_socket, false);
        self::$reg_conn = new TcpConnection($reg_socket, $client_addr);
        self::$reg_conn->protocol = self::$reg_protocol;
        self::$reg_conn->onMessage= [self::class,"onClientTcpMsg"];
        self::$reg_conn->onClose= [self::class,"onClientTcpClose"];
        
        $event = \Workerman\Worker::$globalEvent;
        $event->add(1, EventInterface::EV_TIMER, function () {
            self::$reg_conn->send(['ping']);
        });
        
        $msg_socket = stream_socket_server("udp://$client_addr", $errno, $errstr, STREAM_SERVER_BIND );
        $msg_conn = new \xlx\Connection\ListenUdpConnection($msg_socket, $client_addr);
        $msg_conn->protocol = self::$reg_protocol;
        $msg_conn->onMessage = [self::class,"onClientUdpMsg"];

        \xlx\PrActor::$local_addr = $client_addr;
        echo "CONN:".$client_addr;
        echo "\n";
    }
    //static $server_data=[ 'conn'=>[] ,'link'=>[] ,'nreg'=>[] ];//conn:计算节点 link:绑定
    static $server_conn=[];
    static $server_monitor=[];
    static $server_register=[];
    public static function startServer($event = null)
    {
        if (empty(\Workerman\Worker::$globalEvent)) {
            \Workerman\Worker::$globalEvent = \xlx\Process::createEventLoop();
            $GLOBALS['_']=new class{
                function __destruct(){
                    \Workerman\Worker::$globalEvent->loop();
                }
            };
        }
        if (\xlx\Process::checkAddr('tcp://'.self::$reg_addr)) {
            exit("server started\n");
        } 
        $event = \Workerman\Worker::$globalEvent;
        $socket = \stream_socket_server('tcp://'.self::$reg_addr, $errno, $errstr); 
        $event->add($socket, EventInterface::EV_READ, function ($socket) {
            echo "\nconn\n";
            $socket = \stream_socket_accept($socket, 0, $addr);
            $conn = new TcpConnection($socket, $addr);
            $conn->protocol = self::$reg_protocol;
            $conn->onMessage= [self::class,"onServerTcpMsg"];
            $conn->onClose= [self::class,"onServerTcpClose"];
            self::$server_conn[$addr] = $conn; 
            self::$server_conn[$addr]->addr=$addr;
            
            foreach (self::$server_conn as $addr => $conn) {
                $data[] = $addr;
            }
            foreach (self::$server_conn as $addr => $conn) {
                //$data1 = array_diff($data,[$conn->addr]);
                //print_r($data1);
                $conn->send(['iptable',$data]);
            }
        });
        //iptable
        $event->add(1, EventInterface::EV_TIMER, function () {
            if (count(self::$server_conn)<4) {
                //创建最少一个节点进程
                \xlx\Process::fork(function () {
                    \xlx\PrService::connectServer();
                });
            }

            // file_put_contents('nil',print_r([
            //     'server_conn'=>self::$server_conn,
            //     'server_monitor'=>self::$server_monitor,
            //     'server_register'=>self::$server_register, 
            //     'client_monitor'=>self::$client_monitor,
            //     'client_exit'=>self::$client_exit,
            //     'local_actors'=>\xlx\PrActor::$local_actors
            // ],true));


        });
        
        
        $msg_socket = stream_socket_server("udp://".self::$reg_addr, $errno, $errstr, STREAM_SERVER_BIND );
        $msg_conn = new \xlx\Connection\ListenUdpConnection($msg_socket, self::$reg_addr);
        $msg_conn->protocol = self::$reg_protocol;
        $msg_conn->onMessage = [self::class,"onClientUdpMsg"];

        echo "[START]:".'tcp://'.self::$reg_addr."\n";
    }


    /////////////// server trigger //////////////////

    public static function onServerTcpMsg($conn, $data)
    {
        switch ($data[0]) {
            case 'ping':
                break;
            case 'monitor'://0:TOKEN 1:DEATH 2:TELL 3:RETURN_IP 4:RETURN_TICK 
                self::$server_monitor[$data[1]][$data[2]]=true;     
                break;
            case 'demonitor':
                unset(self::$server_monitor[$data[1]][$data[2]]);
                break; 
            case 'exiting'://0:TOKEN 1:EXIT_COID //die notify
                $links = self::$server_monitor[$data[1]];
                unset(self::$server_monitor[$data[1]]);
                foreach ($links as $key => $_) {    
                    unset(self::$server_monitor[$key][$data[1]]);
                    \xlx\PrActor::exit($key,$data[1]);  
                }
                break; 
            default:
                # code...
                break;
        }
    }
    public static function onServerTcpClose($conn)
    {
        foreach (self::$server_monitor as $coid => $links) { 
            preg_match('/([\d\.]+:\d+):(\w+)/',$coid,$arr);
            if($arr[1]==$conn->addr){
                unset(self::$server_monitor[$coid]);
                foreach ($links as $key => $_) {   
                    \xlx\PrActor::exit($key); 
                }
            }
        }
        echo "CONN DIE\n";
        unset(self::$server_conn[$conn->id]);
        //exit;
    }

    /////////////  client trigger //////////////

    public static function onClientTcpMsg($conn, $data)
    {  
        switch ($data[0]) {
            case 'iptable':
                self::$client_data['iptable'] = $data[1];
                break;  
            default:
                //self::onClientUdpMsg($conn,$data);
                break;
        } 
    }
    public static function onClientTcpClose($conn)
    {
        echo "close:";
        echo \xlx\PrActor::$local_addr;
        echo "\n";
        exit(250);
    }
    public static function onClientUdpMsg($conn, $data)
    {        
        //echo "onClientUdpMsg:";print_r($data);
        switch ($data[0]) {
            case 'send': //0:TOKEN 1:COID 2:MSG
                \xlx\PrActor::send($data[1], $data[2]);
                break;
            case 'exit'://0:TOKEN 1:coid  2:reason_from
                \xlx\PrActor::exit($data[1],$data[2]);
                break;
            case 'spawn'://0:TOKEN 1:FN 2:ARGS 3:OPT 4:RETURN_IP 5:RETURN_TICK 
                unset($data[3]['ip']);//BUG
                \xlx\co(function()use($data){
                    $coid = yield \xlx\PrActor::spawn_opt($data[1],$data[2],$data[3]);
                    self::clientCase($data[4], $a = ['return',$data[5],$coid]);  
                });
                // \xlx\PrActor::spawn_opt($data[1],$data[2],$data[3])->then(function($coid)use($data){ 
                //     self::clientCase($data[4], $a = ['return',$data[5],$coid]);  
                // }); 
                break;
            case 'return'://0:TOKEN 1:TICK 2:RETURN_VALUE  
                if (isset( self::$client_data['return'][$data[1]] )) {
                    self::$client_data['return'][$data[1]]->resolve($data[2]);
                    unset(self::$client_data['return'][$data[1]]);
                } else {   
                    throw new \Exception("Error Processing Request", 1);
                } 
                break; 
            case 'monitor'://0:TOKEN 1:DEATH 2:NOTIFY  3:RETURN_IP 4:RETURN_TICK 
                if(isset(\xlx\PrActor::$local_actors[$data[1]])){
                    self::$client_monitor[$data[1]][$data[2]]=true; 
                    self::serverCase(['monitor',$data[1],$data[2]]); 
                    self::clientCase($data[3],['return',$data[4],'ok']);  
                }else{ 
                    self::clientCase($data[3],['return',$data[4],'err']); 
                }
                break;
            case 'demonitor'://0:TOKEN 1:DEATH 2:NOTIFY  3:RETURN_IP 4:RETURN_TICK 
                unset(self::$client_monitor[$data[1]][$data[2]]); 
                self::serverCase(['demonitor',$data[1],$data[2]]); 
                self::clientCase($data[3],['return',$data[4],'ok']);  
                break;
            default:
                # code...
                break;
        }
    }

    ////////////// method //////////////////////

 
    public static function clientCase($addr, $data)
    {
        if(empty(self::$client_data['socket'][$addr])){
            self::$client_data['socket'][$addr] = stream_socket_client("udp://$addr", $errno, $errstr);
        }
        \fwrite(self::$client_data['socket'][$addr] , self::$reg_protocol::encode($data)); 
    }
 
    public static $tick=0;
    public static function clientCall($addr, $data)
    { 
        $data[] = \xlx\PrActor::$local_addr;
        $data[] = ++self::$tick; 
        self::clientCase($addr, $data);
        self::$client_data['return'][self::$tick] = new \Hprose\Deferred();
        return self::$client_data['return'][self::$tick]->promise;//();
    }
    public static function serverCase($data){
        \xlx\PrService::$reg_conn->send($data);         
    }
    public static function serverCall($data){ 
 
        $data[] = \xlx\PrActor::$local_addr;
        $data[] = ++self::$tick; 
        \xlx\PrService::$reg_conn->send($data); 
        self::$client_data['return'][self::$tick] = new \Hprose\Deferred();
        return self::$client_data['return'][self::$tick]->promise;//();
    }
}
